package com.jacky.iot.mqtt;

import com.jacky.iot.callback.ClientCallback;
import com.jacky.iot.config.MqttClientProperties;
import com.jacky.iot.mediators.ReconnectEvent;
import com.jacky.iot.persist.MqttPersistenceFactory;
import org.eclipse.paho.client.mqttv3.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.event.EventListener;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;


@Component
public class ConnectionManager {
    private static final Logger log = LoggerFactory.getLogger(ConnectionManager.class);
    private final MqttPersistenceFactory persistenceFactory;
    private MqttClient mqttClient;
    private final MqttClientProperties properties;
    private final ClientCallback callback;
    private final AtomicBoolean reconnect = new AtomicBoolean(false);

    public ConnectionManager(MqttClientProperties properties, MqttPersistenceFactory persistenceFactory,
                             ClientCallback callback) {
        this.properties = properties;
        this.persistenceFactory = persistenceFactory;
        this.callback = callback;
        this.initClient();
    }

    public void subscribeTopic(String topic, int qos) throws MqttException {
        if (checkClient()) {
            this.mqttClient.subscribe(topic, qos);
        }
        // todo throw error
    }

    public void unsubscribe(String topic) throws MqttException {
        if (checkClient()) {
            this.mqttClient.unsubscribe(topic);
        }
    }

    public void disconnect() throws MqttException {
        if (checkClient()) {
            this.mqttClient.close();
        }
    }

    public void publish(String topic, String payload, int qos, boolean dup) throws MqttException {
        if (checkClient()) {
            this.mqttClient.publish(topic, payload.getBytes(StandardCharsets.UTF_8), qos, dup);
        }
    }

    private synchronized void initClient() {
        try {
            if (Objects.isNull(mqttClient)) {
                mqttClient = new MqttClient(this.properties.getBroker(), this.properties.getClientId(),
                        this.persistenceFactory.clientPersistence());
                mqttClient.setCallback(this.callback);
                mqttClient.connect(initConnectOptions());
                mqttClient.subscribe(this.properties.getSubTopics()); // 订阅主题
            }
        } catch (MqttSecurityException e) {
            log.error("mqtt client security exception: ", e);
        } catch (MqttException e) {
            log.error("mqtt client conn exception:", e);
        }
    }

    private MqttConnectOptions initConnectOptions() {
        MqttConnectOptions options = new MqttConnectOptions();
        options.setCleanSession(this.properties.getCleanSession());
        options.setUserName(this.properties.getUsername());
        if (Objects.nonNull(this.properties.getPasswd())) {
            options.setPassword(this.properties.getPasswd().toCharArray());
        }
        options.setKeepAliveInterval(this.properties.getKeepAliveInterval()); // 心跳时间
        options.setConnectionTimeout(this.properties.getConnectionTimeout());

        if (Objects.nonNull(this.properties.getWill())) {
            options.setWill(this.properties.getWill().getTopic(), this.properties.getWill().getPayload().getBytes(StandardCharsets.UTF_8),
                    this.properties.getWill().getQos(), this.properties.getWill().getRetained());
        }
        // 如果需要支持SSL,可以在这里配置
//        options.setSocketFactory(SSLSocketFactory);
        return options;
    }

    private boolean checkClient() {
        return Objects.nonNull(this.mqttClient) && this.mqttClient.isConnected();
    }

    @EventListener
    public void reconnect(ReconnectEvent event) {
        log.info("mqtt client reconnect...");
        if (mqttClient != null && !mqttClient.isConnected()) {
            try {
                TimeUnit.SECONDS.sleep(4);
                this.mqttClient = null;
                this.initClient();
            } catch (Exception e) {
                log.error("mqtt client reconnect error:", e);
            }
        }
//        reconnect.set(true);
    }
}
